package com.ali.ttt.netty.cluster.listener;

import com.ali.ttt.core.cocurrent.FutureTaskScheduler;
import com.ali.ttt.netty.client.config.NettyClient;
import com.ali.ttt.netty.cluster.config.NettyClusterProperties;
import com.ali.ttt.netty.cluster.domain.ClientList;
import com.ali.ttt.netty.common.domain.*;
import com.ali.ttt.netty.common.intface.InMessage;
import com.ali.ttt.netty.common.utils.MessageUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.atomic.AtomicReference;


/**
 * @author yuxl
 */
@Slf4j
@Component
public class ClusterListener {

    @Resource
    private NettyClient nettyClient;

    @Resource
    private RedisTemplate<String, Client> redisTemplate;

    @Resource
    private NettyClusterProperties nettyClusterProperties;


    @InMessage(path="/cluster/resLogin",name = "登录响应")
    public void resLogin(Message<LoginRes> req) throws UnknownHostException {

    }

    @InMessage(path="/cluster/login",name = "处理客户端登录")
    public void login(Message<Client> req) throws UnknownHostException {
        Client c = req.getMsg();
        log.info("接收到客户端=>host:{},prod:{} 登录",c.getHost(),c.getProd());
        if(!ClientList.inst().hasConnect(c)){
            Client own = new Client(InetAddress.getLocalHost().getHostAddress(),nettyClusterProperties.getProd());
            AtomicReference<Channel> channel = new AtomicReference<>();
            GenericFutureListener<ChannelFuture> closeListener = (ChannelFuture f) ->
            {
                log.info("连接已经断开……");
                channel.set(f.channel());
            };
            GenericFutureListener<ChannelFuture> connectedListener= (ChannelFuture f) ->
            {
                if (f.isSuccess()){
                    log.info("\n----------------------------------------------------------\n\t" +
                                    "netty客户端 连接成功\n\t" +
                                    "External: \t\033[1;34m{}:{}\033[0m\n\t" +
                                    "----------------------------------------------------------",
                            c.getHost(),
                            c.getProd()
                    );
                    channel.set(f.channel());
                    channel.get().closeFuture().addListener(closeListener);
                    Message<Client> online = new Message<>();
                    online.setPath("/cluster/login");
                    online.setMsg(own);
                    MessageUtil.send(channel.get(),online);
                } else {
                    log.info("连接失败! ");
                    // 清除注册信息
                    redisTemplate.setValueSerializer(RedisSerializer.java());
                    redisTemplate.opsForList().remove(nettyClusterProperties.getKey(),1,c);
                }
            };
            FutureTaskScheduler.add(() ->
            {
                // 反向注册回去
                nettyClient.setHost(c.getHost());
                nettyClient.setPort(c.getProd());
                nettyClient.setCloseListener(closeListener);
                nettyClient.setConnectedListener(connectedListener);
                nettyClient.doConnect();
                ClientList.inst().addServer(c);
            });
        }
    }
}
